package org.wejob.web.buffer;

import org.wejob.web.api.QueueNode;
import org.wejob.web.api.ReadWriteQueue;

import java.util.concurrent.atomic.AtomicInteger;

public abstract class AsynchronousQueue<N extends QueueNode> implements ReadWriteQueue<N> {
    //异步读写队列实现原理；
    /*
     * 当队列中的元素个数>1时，读线程和写线程可以同时进行，因为这时候不涉及操作共享变量
     *当队列中的元素个数<=1时，读写队列中只能有一个线程操作读或者写，因为此时会涉及队列头尾指针的操作；
     * 实现原理，写线程在获取写锁时可以正常做写操作：此时有两种情况--
     *     1，获取写锁之后队列为空，此时不会有读线程做读操作，只有获得写锁的该线程可以put，put完成之后将头尾指针同时指向改为以元素即可；此时队列元素个数为1；
     *     2，获取写锁之后队列中只有一个元素，这时也可以保证只有该线程在做写入，因为只有一个元素的情况下，读线程要读取该元素必须同时获得读锁和写锁；此时队列元素个数为2；
     *     3，读线程获取读锁之后有三种情况；size>1;size=1;size=0;
     *     4, 重点是保证不能多个线程同时进入队列元素为零的状态；就是读线程消费了最后一个元素，正好此时写线程在队列为空的时候写入，读写线程会同时操作头尾指针，造成错乱，所以在元素数量为1
     * 的时候就要进行同步操作；原理：
     *           1.读线程获取读锁之后如果size=1，此时不会先消费，而是试图获取写锁，防止此时有写线程同时操作，获取写锁之后再判断size是否为1，如果为1则做出队操作，然后释放写锁，如果为2则直接释放写锁--再进行出队操作；
     *           2，这里读线程获取读锁之后判断size=1，再获取读锁成功之后有两种情况--
     *                   1，有写线程在读线程之前获取到了写锁，则读线程获取到写锁的时候size>=2了(可能不止一个)，
     *                   2，判断size=1之后直接获取到了写锁，此时就应该阻塞其他写线程做入队操作，等待自己完成出队操作之后再释放写锁；
     *     5，再说一下size怎么保证同步，
     *           1，在size<=1的时候严格保证线程同步操作，保证size；
     *           2，在size>1的时候，此时可以理解为队列同时在出队和入队，size在两个线程操作的时候先出队-1还是先入队+1其实是没有关系的，因为原子操作保证了最后结果是没有问题的就行；
     * */
    protected AtomicInteger size;
    protected N head;
    protected N tail;
    Object readLock;
    Object writeLock;
    //todo 这里考虑使用cas还是Synchronized--cas可以不阻塞读写，获取锁之前可以继续读写数据


    public AsynchronousQueue() {
        this.init();
    }

    //空队列初始化要创建一个node
    AsynchronousQueue(N node) {
        this();
        this.head = node;
        this.tail = this.head;
        this.size.set(1);
    }

    private boolean offerFirstOne(N node) {
        if (this.size.get() > 0) {
            return false;
        }
        this.head = this.tail = node;
        return this.size.compareAndSet(0, 1);
    }

    public boolean offer(N node) {
        preOfferElement(node);
        synchronized (this.writeLock) {
            if (this.size.get() == 0) {
                return this.offerFirstOne(node);
            } else {
                //todo offerOne
                this.head.setPre(node);
                node.setNext(this.head);
                this.head = node;
            }
            return this.increamentSize() > 1;
        }
    }

    void preOfferElement(N bufferNode) {
        bufferNode.setNext(null);
        bufferNode.setPre(null);
    }

    N pollLastOne() {
        return this.size.compareAndSet(1, 0) ? this.tail : null;
    }

    public N poll() {
        synchronized (this.readLock) {
            if (this.size.get() == 0) {
                return getWhileIsEmpty();
            }
            if (this.size.get() == 1) {
                synchronized (this.writeLock) {
                    if (this.size() > 1) {
                        return this.getTailElement();
                    }
                    if (this.size() == 1) {
                        return this.pollLastOne();
                    }
                }
            }
            return this.getTailElement();
        }
    }

    N getTailElement() {
        if (this.size() > 1) {
            this.tail = (N) this.tail.pre();
            this.size.decrementAndGet();
            return (N) this.tail.next();
        }
        return null;
    }

    protected abstract N getWhileIsEmpty();

    public int size() {
        return this.size.get();
    }

    public int increamentSize() {
        return this.size.incrementAndGet();
    }

    public int decrementSize() {
        return this.size.decrementAndGet();
    }

    void init() {
        this.writeLock = new Object();
        this.readLock = new Object();
        this.head = this.tail = null;
        this.size = new AtomicInteger(0);
    }

}

